#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <mysql/mysql.h>
#include "threadpool.h"
#include "communication_utils.h"
#include "cmysql.h"
/*
 * 功能：登入验证
 * 参数：*/
int signin(message *ser, struct threadpool *pool);
void *fun(void *arg);
void thefunofequipment(struct threadpool *pool, struct queue_data data);
void thefunofsensor(struct threadpool *pool, struct queue_data data);
void thefunofadmins(struct threadpool *pool, struct queue_data data);
void inform(message ser, struct threadpool *pool);
void cheak(message ser, int listenfd, struct threadpool *pool);

int main()
{
    int sock = getsock();
    if (sock == -1)
    {
        perror("Error in creating socket");
        return -1;
    }
    int listenfd = 0;
    long id = 0;
    struct threadpool *pool = create_threadpool(10, 10, fun);
    if (!pool)
    {
        perror("Error in creating threadpool");
        close(sock);
        close(listenfd);
        return -1;
    }
    fd_set readfds;
    struct timeval timeout = {10, 0};
    struct queue_data data;
    message ser;
    memset(&ser, 0, sizeof(ser));
    int ret;
    while (pool->runflag)
    {
        FD_ZERO(&readfds);
        FD_SET(sock, &readfds);
        timeout.tv_sec = 10;
        ret = select(sock + 1, &readfds, NULL, NULL, &timeout);
        if (ret == -1)
        {
            perror("Error in select");
            break;
        }
        else if (ret == 0)
        {
            continue;
        }
        listenfd = accept(sock, NULL, NULL);
        printf("New client connected\n");
        ret = recvmessage(listenfd, &ser);
        if (ret == 0)
        {
            perror("Error in receiving message");
            break;
        }
        cheak(ser, listenfd, pool);
        ret = signin(&ser, pool);
        
        while (ret == 0)
        {
            printf("Invalid sign in\n");
            if (ser.O_type == sign_out)
            {
                close(listenfd);
                break;
            }
            if (sendmessage(listenfd, &ser) == 0)
            {
                perror("Error in sending message");
                break;
            }
            if (recvmessage(listenfd, &ser) == 0)
            {
                perror("Error in receiving message");
                break;
            }
            cheak(ser, listenfd, pool);
            ret = signin(&ser, pool);
            
        }
        if (ser.O_type == sign_out)
            continue;
        data.listenfd = listenfd;
        data.d_type = ser.D_type;
        sem_wait(&pool->semw);
        pool->numofcit += 1;
        push(&data, pool->que);
        sem_post(&pool->semr);
    }
    printf("Server exit\n");
    close(sock);
    destroy_threadpool(pool);
    return 0;
}

int signin(message *ser, struct threadpool *pool)
{
    if (ser->O_type != sign_in)
    {
        strcpy(ser->seforms, "非法操作");
        return 0;
    }
    printf("signin\n");
    MYSQL_RES *res = NULL;
    int is_registered = 0;
    switch (ser->D_type)
    {
    case sensor:
        res = resultofquery(pool->conn, "sensor", ser->seforms);
        if (mysql_fetch_row(res) != NULL)
        {
            is_registered = 1;
        }
        strcpy(ser->seforms, "传感器不存在");
        break;
    case equipment:
        res = resultofquery(pool->conn, "equipment", ser->seforms);
        if (mysql_fetch_row(res) != NULL)
        {
            is_registered = 1;
        }
        strcpy(ser->seforms, "设备不存在");
        break;
    case admins:
        res = resultofquery(pool->conn, "admins", ser->seforms);
        if (mysql_fetch_row(res) != NULL)
        {
            is_registered = 1;
        }
        strcpy(ser->seforms, "账号或密码错误");
        break;
    default:
        strcpy(ser->seforms, "非法设备");
        break;
    }
    if (res)
        mysql_free_result(res);
    return is_registered;
}

void *fun(void *arg)
{
    struct threadpool *pool = (struct threadpool *)arg;
    struct queue_data data;
    while(pool->runflag)
    {
        sem_wait(&pool->semr);
        if(!pool->runflag)break;
        pthread_mutex_lock(&pool->mutex);
        data = front(pool->que);
        pop(pool->que);
        pthread_mutex_unlock(&pool->mutex);
        sem_post(&pool->semw);
        switch (data.d_type)
        {
        case sensor:
            thefunofsensor(pool, data);
            pool->sensor_id[data.listenfd] = -1;
            break;
        case equipment:
            thefunofequipment(pool, data);
            pool->equipment_id[data.listenfd] = -1;
            break;
        case admins:
            thefunofadmins(pool, data);
            pool->admin_id[data.listenfd] = -1;
            break;
        default:
            break;
        }
        close(data.listenfd);
        pool->numofcit -= 1;
    }
    return NULL;
}

void thefunofsensor(struct threadpool *pool, struct queue_data data)
{
    sendresult(data.listenfd, "欢迎");
    double temperature, CO2, NO2, CH3COCH3, O3, SO2, C2H2, PM25, humidity;
    char str[256];
    int flag = 1;
    message ser;
    ser.O_type = remessage;
    while (flag && pool->runflag)
    {
        if (recvmessage(data.listenfd, &ser) == 0)
        {
            perror("Error in receiving message");
            break;
        }
        if (ser.O_type == sign_out)
        {
            flag = 0;
            continue;
        }
        sscanf(ser.seforms, "Temperature = %lf, CO2 = %lf, NO2 = %lf, CH3COCH3 = %lf, O3 = %lf, SO2 = %lf, C2H2 = %lf, PM25 = %lf, Humidity = %lf", &temperature, &CO2, &NO2, &CH3COCH3, &O3, &SO2, &C2H2, &PM25, &humidity);
        if (CO2 > 150 || C2H2 > 100 || CH3COCH3 > 100 || humidity > 70 || NO2 > 100 || O3 > 100 || PM25 > 100 || SO2 > 100 || temperature > 50)
        {
            ser.O_type = warring;
            strcpy(ser.seforms, "数据异常");
            inform(ser, pool);
        }
        sprintf(str, "%ld,%.2f, %.2f, %.2f, %.2f, %.2f, %.2f, %.2f, %.2f, %.2f, now()", pool->sensor_id[data.listenfd], temperature, CO2, NO2, CH3COCH3, O3, SO2, C2H2, PM25, humidity);
        insertdata(pool->conn, "sensor_data", "sensor_ID,temperature,humidity,CO2,NO2,CH3COCH3,O3,SO2,C2H2,PM25,time", str);
        if (sendresult(data.listenfd, "数据上传成功") == 0)
        {
            perror("Error in sending result");
            break;
        }
    }
}

void thefunofequipment(struct threadpool *pool, struct queue_data data)
{
    sendresult(data.listenfd, "欢迎");
    message ser;
    ser.O_type = remessage;
    int flag = 1;
    while (flag && pool->runflag)
    {
        sendmessage(data.listenfd, &ser);
        recvmessage(data.listenfd, &ser);
        if (ser.O_type == sign_out)
        {
            flag = 0;
            continue;
        }
        sleep(10);
    }
}

void thefunofadmins(struct threadpool *pool, struct queue_data data)
{
    sendresult(data.listenfd, "欢迎");
    message ser;
    ser.O_type = remessage;
    int flag = 1;
    while (flag && pool->runflag)
    {
        recvmessage(data.listenfd, &ser);
        if(ser.O_type == sign_out)
        {
            flag = 0;
            continue;
        }
        switch (ser.D_type)
        {
        case sensor:
            managesensor(data.listenfd, ser, pool);
            break;
        case equipment:
            memagequipment(data.listenfd, ser, pool);
            break;
        case admins:
            manageadmins(data.listenfd, ser, pool);
            break;
        case server:
            manageserver(data.listenfd, ser, pool);
            break;
        default:
            break;
        }
    }
}
void managesensor(int listenfd, message ser, struct threadpool *pool)
{
    MYSQL_RES *res = NULL;
    MYSQL_ROW row;
    char str[256];
    int flag = 1;
    int id;
    int i;
    switch (ser.O_type)
    {
    case searchonline:
        for (int i = 0; i < 1024; i++)
        {
            if (pool->sensor_id[i] != -1)
            {
                sprintf(str, "传感器ID:%ld", pool->sensor_id[i]);
                sendata(listenfd, str);
            }
        }
        ser.O_type = stopdata;
        sendmessage(listenfd, &ser);
        break;
    case addid:
        insertdata(pool->conn, "sensor", "sensor_ID", ser.seforms);
        strcpy(ser.seforms, "添加传感器成功");
        sendresult(listenfd, ser.seforms);
        break;
    case updataid:
        updatedata(pool->conn, "sensor", ser.seforms);
        strcpy(ser.seforms, "更新传感器信息成功");
        sendresult(listenfd, ser.seforms);
        break;
    case delid:
        deletedata(pool->conn, "sensor", ser.seforms);
        strcpy(ser.seforms, "删除传感器成功");
        sendresult(listenfd, ser.seforms);
        break;
    case closeid:
        sscanf(ser.seforms, "sensor_ID = %4ld", &id);
        for (i = 0; i < 1024; i++)
        {
            if (pool->sensor_id[i] == id)
                break;
        }
        if (i == 1024)
        {
            strcpy(ser.seforms, "传感器不在线");
            sendresult(listenfd, ser.seforms);
        }
        else
        {
            sendexit(i);
            strcpy(ser.seforms, "关闭传感器成功");
            sendresult(listenfd, ser.seforms);
        }
        break;
    case searchdata:
        res =resultofquery(pool->conn, "sensor_data", ser.seforms);
        while ((row = mysql_fetch_row(res))!= NULL)
        {
            sprintf(str, "temperature:%s, humidity:%s, CO2:%s, NO2:%s, CH3COCH3:%s, O3:%s, SO2:%s, C2H2:%s, PM25:%s, time:%s",
                    row[1], row[2], row[3], row[4], row[5], row[6], row[7], row[8], row[9], row[10]);
            sendata(listenfd, str);
        }
        mysql_free_result(res);
        ser.O_type = stopdata;
        sendmessage(listenfd, &ser);
        break;
    default:
        break;
    }
}
void memagequipment(int listenfd, message ser, struct threadpool *pool)
{
    MYSQL_RES *res = NULL;
    MYSQL_ROW row;
    char str[256];
    int flag = 1;
    int id;
    int i;
    switch (ser.O_type)
    {
    case searchonline:
        for (int i = 0; i < 1024; i++)
        {
            if (pool->equipment_id[i] != -1)
            {
                sprintf(str, "设备ID:%ld", pool->equipment_id[i]);
                sendata(listenfd, str);
            }
        }
        ser.O_type = stopdata;
        sendmessage(listenfd, &ser);
        break;
    case addid:
        insertdata(pool->conn, "equipment", "equipment_ID", ser.seforms);
        strcpy(ser.seforms, "添加设备成功");
        sendresult(listenfd, ser.seforms);
        break;
    case updataid:
        updatedata(pool->conn, "equipment", ser.seforms);
        strcpy(ser.seforms, "更新设备信息成功");
        sendresult(listenfd, ser.seforms);
        break;
    case delid:
        deletedata(pool->conn, "equipment", ser.seforms);
        strcpy(ser.seforms, "删除设备成功");
        sendresult(listenfd, ser.seforms);
        break;
    case closeid:
        sscanf(ser.seforms, "equipment_ID = %4ld", &id);
        for (i = 0; i < 1024; i++)
        {
            if (pool->equipment_id[i] == id)
                break;
        }
        if (i == 1024)
        {
            strcpy(ser.seforms, "设备不在线");
            sendresult(listenfd, ser.seforms);
        }
        else
        {
            sendexit(i);
            strcpy(ser.seforms, "关闭设备成功");
            sendresult(listenfd, ser.seforms);
        }
        break;
    default:
        break;
    }
}
void manageadmins(int listenfd, message ser, struct threadpool *pool)
{
    MYSQL_RES *res = NULL;
    MYSQL_ROW row;
    char str[256];
    int flag = 1;
    int id;
    int i;
    switch (ser.O_type)
    {
    case searchonline:
        for (int i = 0; i < 1024; i++)
        {
            if (pool->admin_id[i] != -1)
            {
                sprintf(str, "管理员ID:%ld", pool->admin_id[i]);
                sendata(listenfd, str);
            }
        }
        ser.O_type = stopdata;
        sendmessage(listenfd, &ser);
        break;
    case addid:
        insertdata(pool->conn, "admins", "account,password", ser.seforms);
        strcpy(ser.seforms, "添加管理员成功");
        sendresult(listenfd, ser.seforms);
        break;
    case updataid:
        updatedata(pool->conn, "admins", ser.seforms);
        strcpy(ser.seforms, "更新管理员信息成功");
        sendresult(listenfd, ser.seforms);
        break;
    case delid:
        deletedata(pool->conn, "admins", ser.seforms);
        strcpy(ser.seforms, "删除管理员成功");
        sendresult(listenfd, ser.seforms);
        break;
    case closeid:
        sscanf(ser.seforms, "account = %13ld and password = '%s'", &id, ser.seforms);
        for (i = 0; i < 1024; i++)
        {
            if (pool->admin_id[i] == id)
                break;
        }
        if (i == 1024)
        {
            strcpy(ser.seforms, "管理员不在线");
            sendresult(listenfd, ser.seforms);
        }
        else
        {
            sendexit(i);
            strcpy(ser.seforms, "关闭管理员成功");
            sendresult(listenfd, ser.seforms);
        }
        break;
    case sign_out:
        sendexit(listenfd);
        break;
    default:
        break;
    }
}
void manageserver(int listenfd, message ser, struct threadpool *pool)
{
    switch (ser.O_type)
    {
    case searchdata:
        sprintf(ser.seforms, "服务器线程数:%d, 当前连接数:%d", pool->threads, pool->numofcit);
        sendresult(listenfd, ser.seforms);
        break;
    case closeid:
        pool->runflag = 0;
        ser.O_type = sign_out;
        ser.D_type = server;
        strcpy(ser.seforms, "exit");
        inform(ser, pool);
        break;
    case quit:
        sendexit(listenfd);
        break;
    default:
        break;
    }
}
void inform(message ser, struct threadpool *pool)
{
    for(int i = 0; i < 1024; i++)
    {
        if (pool->equipment_id[i] != -1)
        {
            sendmessage(i, &ser);
        }
        if (pool->admin_id[i] != -1)
        {
            sendmessage(i, &ser);
        }
        if (pool->sensor_id[i] != -1)
        {
            sendmessage(i, &ser);
        }
    }
}
void cheak(message ser, int listenfd, struct threadpool *pool)
{
    char str[256];
    switch (ser.D_type)
    {
    case sensor:
        printf("%s\n", ser.seforms);
        sscanf(ser.seforms, "sensor_ID = %4ld", &pool->sensor_id[listenfd]);
        printf("sensor_ID:%ld\n", pool->sensor_id[listenfd]);
        break;
    case equipment:
        sscanf(ser.seforms, "equipment_ID = %4ld", &pool->equipment_id[listenfd]);
        break;
    case admins:
        sscanf(ser.seforms, "account = %13ld and password = '%s'", &pool->admin_id[listenfd], str);
        break;
    default:
        break;
    }
}